package com.ztorn.driver;

import cn.hutool.core.text.CharSequenceUtil;
import com.ztorn.common.assertion.Asserts;
import com.ztorn.common.model.Column;
import com.ztorn.common.model.ProcessEntity;
import com.ztorn.common.model.Table;
import com.ztorn.common.util.LogUtil;
import com.ztorn.common.util.SqlUtil;
import com.ztorn.convert.DorisTypeConvert;
import com.ztorn.framework.context.ProcessContextHolder;
import com.ztorn.framework.convert.ITypeConvert;
import com.ztorn.framework.driver.AbstractJdbcDriver;
import com.ztorn.framework.query.IDBQuery;
import com.ztorn.framework.result.JdbcSelectResult;
import com.ztorn.query.DorisQuery;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
public class DorisDriver extends AbstractJdbcDriver {

    @Override
    public IDBQuery getDBQuery() {
        return new DorisQuery();
    }

    @Override
    public ITypeConvert getTypeConvert() {
        return new DorisTypeConvert();
    }

    @Override
    public String getDriverClass() {
        return "com.mysql.cj.jdbc.Driver";
    }

    @Override
    public String generateCreateTableSql(Table table) {
        String genTableSql = genTable(table);
        log.info("Auto generateCreateTableSql {}", genTableSql);
        return genTableSql;
    }

    @Override
    public String getType() {
        return "Doris";
    }

    @Override
    public String getName() {
        return "Doris";
    }

    @Override
    public JdbcSelectResult executeSql(String sql, Integer limit) {
        ProcessEntity process = ProcessContextHolder.getProcess();
        process.info("Start parse sql...");
        String[] statements = SqlUtil.getStatements(SqlUtil.removeNote(sql));
        process.info(
                CharSequenceUtil.format(
                        "A total of {} statement have been Parsed.", statements.length));
        List<Object> resList = new ArrayList<>();
        JdbcSelectResult result = JdbcSelectResult.buildResult();
        process.info("Start execute sql...");
        for (String item : statements) {
            String type = item.toUpperCase();
            if (type.startsWith("SELECT") || type.startsWith("SHOW") || type.startsWith("DESC")) {
                process.info("Execute query.");
                result = query(item, limit);
            } else if (type.startsWith("INSERT")
                    || type.startsWith("UPDATE")
                    || type.startsWith("DELETE")) {
                try {
                    process.info("Execute update.");
                    resList.add(executeUpdate(item));
                    result.setStatusList(resList);
                    result.success();
                } catch (Exception e) {
                    resList.add(0);
                    result.setStatusList(resList);
                    result.error(LogUtil.getError(e));
                    process.error(e.getMessage());
                    return result;
                }
            } else {
                try {
                    process.info("Execute DDL.");
                    execute(item);
                    resList.add(1);
                    result.setStatusList(resList);
                    result.success();
                } catch (Exception e) {
                    resList.add(0);
                    result.setStatusList(resList);
                    result.error(LogUtil.getError(e));
                    process.error(e.getMessage());
                    return result;
                }
            }
        }
        return result;
    }

    @Override
    public Map<String, String> getFlinkColumnTypeConversion() {
        HashMap<String, String> map = new HashMap<>();
        map.put("DATETIME", "STRING");
        map.put("TIMESTAMP", "DATETIME");
        return map;
    }

    public String genTable(Table table) {
        String columnStrs =
                table.getColumns().stream()
                        .map(
                                column -> {
                                    String unit = "";
                                    if(column.getScale()!=null && column.getScale()>9){
                                        column.setScale(9);
                                    }
                                    if (column.getPrecision() != null
                                            && column.getScale() != null
                                            && column.getPrecision() > 0
                                            && column.getScale() > 0) {
                                        unit =
                                                String.format(
                                                        "(%s,%s)",
                                                        column.getPrecision(), column.getScale());
                                    } else if (null != column.getLength()) {
                                        unit = String.format("(%s)", column.getLength());
                                    }

                                    final String dv = column.getDefaultValue();
                                    String defaultValue =
                                            Asserts.isNotNull(dv)
                                                    ? String.format(
                                                    " DEFAULT %s",
                                                    "".equals(dv) ? "\"\"" : "\""+dv+"\"")
                                                    : String.format(
                                                    "%s NULL ",
                                                    !column.isNullable() ? " NOT " : "");
                                    if(StringUtils.isNotBlank(dv)) {
                                        if (dv.contains("current_timestamp") || dv.contains("CURRENT_TIMESTAMP")) {
                                            defaultValue = "";
                                        }
                                    }
                                    String columnType = column.getType();
                                    if(getFlinkColumnTypeConversion().containsKey(columnType.toUpperCase())){
                                        columnType = getFlinkColumnTypeConversion().get(columnType.toUpperCase());
                                    }
                                    if(columnType.equalsIgnoreCase("varchar")){
                                        unit = String.format("(%s)", column.getLength()*3);
                                    }
                                    if(columnType.equalsIgnoreCase("json")){
                                        columnType = "String";
                                    }
                                    if(columnType.equalsIgnoreCase("DATE")){
                                        columnType = "DATEV2";
                                    }
                                    if(columnType.contains("tinyint unsigned")){
                                        columnType = "SMALLINT";
                                    }
                                    if(columnType.contains("double")){
                                        columnType = "DECIMAL";
                                    }
                                    if(columnType.contains("mediumtext")){
                                        columnType = "ARRAY<ROW<billNo STRING,collectionId STRING,collectionName STRING>>";
                                        defaultValue = "";
                                    }

                                    return String.format(
                                            "  `%s`  %s%s%s%s%s",
                                            column.getName(),
                                            columnType,
                                            unit,
                                            defaultValue,
                                            "",
                                            Asserts.isNotNullString(column.getComment())
                                                    ? String.format(
                                                    " COMMENT '%s'", column.getComment())
                                                    : "");
                                })
                        .collect(Collectors.joining(",\n"));

        List<String> columnKeys =
                table.getColumns().stream()
                        .filter(Column::isKeyFlag)
                        .map(Column::getName)
                        .map(t -> String.format("`%s`", t))
                        .collect(Collectors.toList());

        String primaryKeyStr =
                columnKeys.isEmpty()
                        ? ""
                        : columnKeys.stream()
                        .collect(Collectors.joining(",", "", ""));

        return MessageFormat.format(
                "CREATE TABLE IF NOT EXISTS `{0}`.`{1}` (\n{2})\n {3}\n{4}\n{5};",
                table.getSchema(),
                table.getName(),
                columnStrs,
                "UNIQUE KEY("+primaryKeyStr+")",
                "DISTRIBUTED BY HASH("+primaryKeyStr+") BUCKETS 1",
                "PROPERTIES (\n" +
                        "\"replication_allocation\" = \"tag.location.default: 1\",\n" +
                        "\"light_schema_change\" = \"true\"\n" +
                        ")");
    }
}
